package com.microservices.yucloud.api.job;

import com.microservices.yucloud.api.entity.Email;
import com.microservices.yucloud.api.service.EmailService;
import com.microservices.yucloud.api.service.WorkerService;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.handler.annotation.JobHandler;
import com.xxl.job.core.log.XxlJobLogger;
import com.xxl.job.core.util.ShardingUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.stream.Stream;

/**
 * 分片广播任务
 *
 * @author xuxueli 2017-07-25 20:56:50
 */
@Slf4j
@JobHandler(value = "shardingJobHandler")
@Service
public class ShardingJobHandler extends IJobHandler {

    @Autowired
    private EmailService emailService;

    @Autowired
    private WorkerService workerService;

    @Override
    public ReturnT<String> execute(String param) throws Exception {

        // 分片参数
        ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
        //执行器数量
        int total = shardingVO.getTotal();
        //当前分片
        int index = shardingVO.getIndex();

        XxlJobLogger.log("分片参数：当前分片序号 = {}, 总分片数 = {}", index, total);

        List<Email> tasks = emailService.selectEmailTasks();

        int count = tasks.size();

        if (tasks != null && count > 0) {
            // 业务逻辑 并行流
            Stream.iterate(0, i -> i + 1).limit(count).parallel().forEach(i -> {
                //根据集群分片平均分配任务
                if (i % total == index) {
                    try {
                        XxlJobLogger.log("第 {} 片, 命中分片开始处理任务:{}", index, i);
                        workerService.sendQueue(tasks.get(i));
                    } catch (Exception e) {
                        log.error("生产消息异常", e);
                    }
                } else {
                    XxlJobLogger.log("第 {} 片, 忽略处理任务:{}", index, i);
                }
            });
        } else {
            log.info("任务空闲");
        }

        return SUCCESS;
    }

}
